package cn.edu360

import java.sql.DriverManager
import java.util.Properties

import com.typesafe.config.ConfigFactory
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object Report {
  def main(args: Array[String]): Unit = {
    if(args.length!=2){
     println( """
        |cn.edu360.Report
        |参数:input output
      """.stripMargin
     )
      sys.exit(1)
    }
    //接收参数
    val Array(input,output)=args
    //conf
    val conf: SparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(this.getClass.getSimpleName)
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc: SparkContext = new SparkContext(conf)
    val sqlcontext: SQLContext = new SQLContext(sc)
    val parquetDate: DataFrame = sqlcontext.read.parquet(input)
    //定义一张临时表
    parquetDate.registerTempTable("t_report")
    //sqlcontext.sql("select * from t_report").show(3)//验证数据
   // sqlcontext.sql("select provincename,cityname, count(*)ct from t_report group by provincename,cityname order by provincename,cityname desc").coalesce(1).write.json(output)
    //把结果写入mysql
   //加载配置文件
   val url="jdbc:mysql://127.0.0.1:3306/wjy1?useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true&zeroDateTimeBehavior=convertToNull&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=UTC&autoReconnect=true&failOverReadOnly=false&useSSL=true"
   val table="t_report"
    val conn: Properties = new Properties()
    conn.setProperty("usr","wjy1")
    conn.setProperty("password","351991")
    conn.setProperty("driver","com.mysql.jdbc.Driver")
  // val config = ConfigFactory.load()
   // val conn = DriverManager.getConnection(config.getString("db.url"), config.getString("db.user"), config.getString("db.password"))
    sqlcontext.sql("select provincename,cityname, count(*)ct from t_report group by provincename,cityname order by provincename,cityname desc").coalesce(1).write.jdbc(url,table,conn)
  }
}
